#include "stream_client.h"

#include "repo.h"
#include "wl_net.h"

#ifdef __cplusplus
extern "C" {
#endif

typedef struct __stream_client_global {
    wl_bool                initialized;
    wl_uint32              state;
    wl_char*               sendbuf;
    wl_int32               sendbufsize;     // 缓存区大小
    wl_int32               sendbufdatalen;  // 缓存区数据长度
    wl_int32               sendbufpos;
    wl_time                latest_recv_timestamp;
    wl_uint32              recv_bytes;
    WL_TIMER               timer;
    wl_ext_tcp_handle      network_handle;
    stream_client_callback callback;
} stream_client_global;

static stream_client_global s_stream_client_global;

static void client_callback(wl_uint32 event, void* data, wl_uint32 datalen) {
    if (wl_null != s_stream_client_global.callback) {
        s_stream_client_global.callback(event, data, datalen);
    }
}

static void reset_connect(void) {
    s_stream_client_global.state          = STREAM_CLIENT_STATE_DISCONNECTED;
    s_stream_client_global.sendbufdatalen = 0;
    s_stream_client_global.sendbufpos     = 0;
    if (WL_INVALID_TIMER != s_stream_client_global.timer) {
        wl_timer_stop(s_stream_client_global.timer);
        s_stream_client_global.timer = WL_INVALID_TIMER;
    }

    if (NULL != s_stream_client_global.network_handle) {
        wl_ext_tcp_disconnect(s_stream_client_global.network_handle);
        s_stream_client_global.network_handle = NULL;
    }
    
}

static wl_int32 process_read_event(void) {
    wl_char   buffer[1024];
    wl_int32  readlen   = 0;
    wl_uint32 recvbytes = 0;
    while (wl_true) {
        readlen = wl_ext_tcp_recv(s_stream_client_global.network_handle, buffer, 1024);
        if (0 < readlen) {
            client_callback(STREAM_CLIENT_EVENT_RECV_DATA, buffer, readlen);
            recvbytes += readlen;
        }else if (0 == readlen) {
            //TODO: waiting for data can read
            WLLOGT("buffer empty");
            break;
        }else {
            WLLOGW("tcp connect invalid");
            reset_connect();
            client_callback(STREAM_CLIENT_EVENT_DISCONNECTED, wl_null, 0);
            return -1;
        }
    }
    s_stream_client_global.recv_bytes += recvbytes;
    s_stream_client_global.latest_recv_timestamp = repo_get_server_timestamp();
    return 0;
}

static wl_int32 process_write_event(void) {
    WLLOGT("sendbufdatalen %d sendbufpos %d", s_stream_client_global.sendbufdatalen, s_stream_client_global.sendbufpos);
    while (s_stream_client_global.sendbufdatalen > 0) {
        wl_int32 datalen = s_stream_client_global.sendbufdatalen > 1024 ? 1024 : s_stream_client_global.sendbufdatalen;
        wl_int32 sendlen = wl_ext_tcp_send(s_stream_client_global.network_handle, &s_stream_client_global.sendbuf[s_stream_client_global.sendbufpos], datalen);
        WLLOGD("sendlen %d", sendlen);
        if (sendlen > 0) {
            s_stream_client_global.sendbufpos += sendlen;
            s_stream_client_global.sendbufdatalen -= sendlen;
            if (0 >= s_stream_client_global.sendbufdatalen) {
                s_stream_client_global.sendbufpos = 0;
                break;
            }
        } else if (0 == sendlen) {
            WLLOGT("buffer full");
            break;
        } else {
            reset_connect();
            client_callback(STREAM_CLIENT_EVENT_DISCONNECTED, wl_null, 0);
            WLLOGW("send exception");
            return -1;
        }
    }
    return 0;
}

static void wl_ext_client_network_callback(wl_ext_tcp_handle handle, int tcpEvent, void* context) {
    WLLOGD("tcpEvent: %d", tcpEvent);
    switch (tcpEvent) {
        case WL_EXT_TCP_CONNECTING: {
        }
        break;

        case WL_EXT_TCP_CONNECTED: {
            if (WL_INVALID_TIMER != s_stream_client_global.timer) {
                wl_timer_stop(s_stream_client_global.timer);
                s_stream_client_global.timer = WL_INVALID_TIMER;
            }
            s_stream_client_global.state = STREAM_CLIENT_STATE_CONNECTED;
            client_callback(STREAM_CLIENT_EVENT_CONNECTED, wl_null, 0);
        }
        break;

        case WL_EXT_TCP_CAN_SEND: {
            process_write_event();
        }
        break;

        case WL_EXT_TCP_CAN_RECV: {
            process_read_event();
        }
        break;

        case WL_EXT_TCP_DISCONNECT:
        case WL_EXT_TCP_ERROR: {
            if (handle == s_stream_client_global.network_handle) {
                reset_connect();
                client_callback(STREAM_CLIENT_EVENT_DISCONNECTED, wl_null, 0);
            }
        }
        break;
    }
}

static void steam_client_timer_callback(WL_TIMER timer, void* context) {
    if (timer != s_stream_client_global.timer) {
        WLLOGW("timer %p is not equal local timer %p", timer, s_stream_client_global.timer);
        wl_timer_stop(timer);
        return;
    }
    reset_connect();
    client_callback(STREAM_CLIENT_EVENT_DISCONNECTED, wl_null, 0);
}

wl_int32 stream_client_init(stream_client_callback callback) {
    if (!s_stream_client_global.initialized) {
        s_stream_client_global.sendbuf = (wl_char*)wl_malloc(WL_MEMORY_STREAM_CACHE_MAX);
        if (wl_null == s_stream_client_global.sendbuf) {
            WLLOGW("wl_malloc fail");
            return -1;
        }

        s_stream_client_global.network_handle = NULL;
        s_stream_client_global.sendbufsize = WL_MEMORY_STREAM_CACHE_MAX;
        reset_connect();
        s_stream_client_global.callback    = callback;
        s_stream_client_global.initialized = wl_true;
    }
    WLLOGI("stream_client_init");
    return 0;
}

wl_int32 stream_client_end(void) {
    if (s_stream_client_global.initialized) {
        if (wl_null != s_stream_client_global.sendbuf) {
            wl_free(s_stream_client_global.sendbuf);
            s_stream_client_global.sendbuf = wl_null;
        }
        s_stream_client_global.sendbufsize = 0;
        reset_connect();
        s_stream_client_global.callback    = wl_null;
        s_stream_client_global.initialized = wl_false;
    }
    WLLOGI("stream_client_end");
    return 0;
}

wl_int32 stream_client_get_state(void) {
    // WLLOGD("state %d", s_stream_client_global.state);
    return s_stream_client_global.state;
}

wl_int32 stream_client_connect(wl_char* server_addr, wl_uint16 server_port, wl_uint32 timeout) {
    reset_connect();

    // 连接超时定时器
    s_stream_client_global.timer =
        wl_timer_start(timeout, NON_REPETITIVE, steam_client_timer_callback, wl_null, WL_TASK_APP, "stream_client");
    if (WL_INVALID_TIMER == s_stream_client_global.timer) {
        WLLOGW("wl_timer_start fail");
        reset_connect();
        return -1;
    }

    s_stream_client_global.network_handle = wl_ext_tcp_connect(server_addr, server_port, wl_ext_network_callback_wrapper, wl_ext_client_network_callback);
    if (NULL == s_stream_client_global.network_handle) {
        WLLOGW("wl_ext_network_connect fail");
        return -1;
    }

    s_stream_client_global.state      = STREAM_CLIENT_STATE_CONNECTING;
    s_stream_client_global.recv_bytes = 0;
    return 0;
}

wl_int32 stream_client_disconnect(void) {
    reset_connect();
    client_callback(STREAM_CLIENT_EVENT_DISCONNECTED, wl_null, 0);
    WLLOGI("stream_client_disconnect");
    return 0;
}

wl_time stream_client_get_latest_recv_timestamp(void) {
    return s_stream_client_global.latest_recv_timestamp;
}

wl_uint32 stream_client_get_recv_bytes(void) {
    return s_stream_client_global.recv_bytes;
}

wl_int32 stream_client_write_data(wl_char* data, wl_uint32 datalen) {
    {
        // 计算剩余缓存大小
        // 缓存区大小-数据大小
        wl_int32 leftsize = s_stream_client_global.sendbufsize - s_stream_client_global.sendbufdatalen;
        if (leftsize < datalen) {
            WLLOGW("leftsize %d is small than datalen %d", leftsize, datalen);
            return -1;
        }
    }
    {
        // 是否需要移动缓存区
        // 计算缓存区尾部剩余大小
        // 缓存区大小-当前发送位置-缓存数据大小
        wl_int32 tailpos  = s_stream_client_global.sendbufpos + s_stream_client_global.sendbufdatalen;
        wl_int32 tailsize = s_stream_client_global.sendbufsize - tailpos;
        // 尾部空间不够,前移数据
        if (tailsize < datalen) {
            wl_int32 datapos = s_stream_client_global.sendbufpos;
            wl_int32 copylen = 0;
            for (; datapos < s_stream_client_global.sendbufsize && copylen < s_stream_client_global.sendbufdatalen;
                 ++datapos, ++copylen) {
                s_stream_client_global.sendbuf[copylen] = s_stream_client_global.sendbuf[datapos + copylen];
            }
            s_stream_client_global.sendbufpos = 0;
        }
        // 尾部追加数据
        tailpos = s_stream_client_global.sendbufpos + s_stream_client_global.sendbufdatalen;
        memcpy(&s_stream_client_global.sendbuf[tailpos], data, datalen);
        s_stream_client_global.sendbufdatalen += datalen;
    }
    process_write_event();
    return 0;
}

#ifdef __cplusplus
}
#endif
